package com.example.mysql.debezium.framework;

import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;

import java.io.Serializable;
import java.util.List;
import java.util.Map;

public interface TableListener {
    String database();

    String table();

    void onWrite(EventType eventType, WriteRowsEventData event);

    void onUpdate(EventType eventType, UpdateRowsEventData event);

    void onDelete(EventType eventType, DeleteRowsEventData event);


    abstract class Base implements TableListener {
        private final String database;
        private final String tableName;

        public Base(String database, String tableName) {
            this.database = database;
            this.tableName = tableName;
        }

        @Override
        public String database() {
            return database;
        }

        @Override
        public String table() {
            return tableName;
        }

        @Override
        public void onWrite(EventType eventType, WriteRowsEventData event) {
            this.onWrite(eventType, event.getRows());
        }

        protected void onWrite(EventType eventType, List<Serializable[]> rows) {
            rows.forEach(i -> this.onUpdateRow(eventType, i));
        }

        protected void onWriteRow(EventType eventType, Serializable[] row) {
            System.err.println(row[0]);
        }

        @Override
        public void onUpdate(EventType eventType, UpdateRowsEventData event) {
            this.onUpdate(eventType, event.getRows().stream().map(Map.Entry::getKey).toList());
        }

        protected void onUpdate(EventType eventType, List<Serializable[]> rows) {
            rows.forEach(i -> this.onUpdateRow(eventType, i));
        }

        protected void onUpdateRow(EventType eventType, Serializable[] row) {
            System.err.println(row[0]);
        }

        @Override
        public void onDelete(EventType eventType, DeleteRowsEventData event) {
            this.onDelete(eventType, event.getRows());
        }

        protected void onDelete(EventType eventType, List<Serializable[]> rows) {
            rows.forEach(i -> this.onDeleteRow(eventType, i));
        }

        protected void onDeleteRow(EventType eventType, Serializable[] row) {
            System.err.println(row[0]);
        }
    }
}
